package org.pupil.socket.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.pupil.common.constants.MsgType;
import org.pupil.common.constants.ProtocolConstants;
import org.pupil.common.constants.RpcSerialization;
import org.pupil.config.Properties;
import org.pupil.socket.compress.Compress;
import org.pupil.socket.compress.CompressFactory;
import org.pupil.socket.serialization.Serialization;
import org.pupil.socket.serialization.SerializationFactory;

import java.util.List;

/**
 * @Description: 解码器
 * @Author: pupil
 * @Date: 2024/07/06 下午 9:55
 */
public class RpcDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 如果可读数据长度小于协议头总长度证明没有
        if (in.readableBytes() < ProtocolConstants.HEADER_TOTAL_LEN) {
            return;
        }

        // 标记当前读取位置，方便后面回退
        in.markReaderIndex();

        // 读取魔数字段
        short magic = in.readShort();
        if (magic != ProtocolConstants.MAGIC) {
            throw new IllegalArgumentException("magic number is illegal, " + magic);
        }
        // 读取版本字段
        byte version = in.readByte();
        // 读取消息类型
        byte msgType = in.readByte();
        // 读取响应状态
        byte status = in.readByte();
        // 读取请求id
        long requestId = in.readLong();
        // 读取消息体长度
        int dataLength = in.readInt();
        // 如果可读字节数小于消息体长度，说明还没有接收完整个消息体，回退并返回
        if (in.readableBytes() < dataLength) {
            // 回退标记位置
            in.resetReaderIndex();
            return;
        }
        // 读取消息体
        byte[] data = new byte[dataLength];
        in.readBytes(data);

        // 处理消息类型
        MsgType msgTypeEnum = MsgType.findByType(msgType);
        if (msgTypeEnum == null) {
            return;
        }

        // 构建消息头
        MsgHeader header = new MsgHeader();
        header.setMagic(magic);
        header.setVersion(version);
        header.setMsgType(msgType);
        header.setStatus(status);
        header.setRequestId(requestId);
        header.setMsgLen(dataLength);

        // 构建消息
        RpcProtocol rpcProtocol = new RpcProtocol<>();
        rpcProtocol.setHeader(header);
        Compress compress = CompressFactory.get(Properties.getCompression());
        Serialization rpcSerialization = SerializationFactory.get(Properties.getSerialization());
        // 反序列化对应的消息类型
        switch (msgTypeEnum){
            case REQUEST:
                data = compress.decompress(data);
                RpcRequest rpcRequest = rpcSerialization.deSerialize(data, RpcRequest.class);
                rpcProtocol.setBody(rpcRequest);
                break;
            case RESPONSE:
                data = compress.decompress(data);
                RpcResponse rpcResponse = rpcSerialization.deSerialize(data, RpcResponse.class);
                rpcProtocol.setBody(rpcResponse);
                break;
        }
        out.add(rpcProtocol);
    }

}
